package com.example.testsignalr

import android.os.AsyncTask
import android.util.Log
import com.microsoft.signalr.HubConnection
import com.microsoft.signalr.HubConnectionBuilder
import com.microsoft.signalr.HubConnectionState
import com.microsoft.signalr.Subscription
import io.reactivex.Completable
import io.reactivex.Observable
import io.reactivex.Single
import io.reactivex.SingleObserver
import io.reactivex.disposables.Disposables
import java.lang.Exception
import java.util.concurrent.TimeUnit
import kotlin.math.pow


class AccessTokenProvider : Single<String>() {
    var authToken: String? = null

    override fun subscribeActual(observer: SingleObserver<in String>) {
        observer.onSubscribe(Disposables.disposed())
        observer.onSuccess(authToken ?: "")
    }
}

open class RxSignalrClient(
    protected open val hubUrl: String
) {
    internal class HubConnectionTask : AsyncTask<HubConnection?, Void?, Void?>() {
        override fun onPreExecute() {
            super.onPreExecute()
        }

        override fun doInBackground(vararg params: HubConnection?): Void? {
            val hubConnection = params[0]
            hubConnection!!.onClosed {
                reconnect(hubConnection)
            }
            reconnect(hubConnection)
            return null
        }

        private fun reconnect(hubConnection:HubConnection) {
            Log.d(TAG,"开始连接")

            var retryCount = 1

            if (hubConnection.connectionState != HubConnectionState.CONNECTED) {
                do {
                    Log.d(TAG,"第${retryCount}次尝试")
                    val waitTimeInit =
                        2.0.pow(retryCount.toDouble()).toLong() * 1000L
                    val waitTime = waitTimeInit.coerceAtMost(MAX_WAIT_INTERVAL)
                    try {
                        Thread.sleep(waitTime)
                        hubConnection.start().blockingAwait(10,TimeUnit.SECONDS)
                    } catch (e: InterruptedException) {
                        // do nothing
                    } catch (e:Exception) {

                    }

                } while (hubConnection.connectionState != HubConnectionState.CONNECTED && retryCount++ < MAX_RETRY)
                Log.d(TAG,"多次尝试后建立了连接")
            } else {
                Log.d(TAG,"连接已经建立")
            }
        }

    }

    open val hubConnection: HubConnection by lazy {
        buildHubConnection()
    }

    companion object {
        const val TAG = "RxSignalrClient"
        const val MAX_WAIT_INTERVAL = 60000L
        const val MAX_RETRY = 60
        const val SERVER_TIMEOUT = 10000L
        const val KEEP_ALIVE_TIME = 10000L
    }

    open fun startConnection() {
        HubConnectionTask().execute(hubConnection)
    }

    open fun stopConnection(): Completable = hubConnection.stop()

    open fun isConnected(): Boolean = hubConnection.connectionState == HubConnectionState.CONNECTED

    inline fun <reified T> subscribeOn(hubMethod: String): Observable<T> {
        var subscription: Subscription? = null

        return Observable.create<T> { emitter ->
            subscription = hubConnection.on(hubMethod, emitter::onNext, T::class.java)
        }
            .doOnDispose { subscription?.unsubscribe() }
    }

    inline fun <reified T> call(hubMethod: String,vararg params:Any):Observable<T> {
        return hubConnection.invoke(T::class.java,hubMethod,params).toObservable()
    }



    open fun send(hubMethod: String?, vararg args: Any?):Boolean {
        if (!isConnected()) {
            return false
        }
        hubConnection.send(hubMethod,*args) //这里一定要使用*号，否则会调用失败
        return true
    }

    protected open fun buildHubConnection(): HubConnection = HubConnectionBuilder.create(hubUrl)
//        .withAccessTokenProvider(accessTokenProvider)
        .build()
}